package ex.datastream.connectors;

import ex.datastream.sources.FileDataSource;
import ex.vo.CityPojo;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 示例：基于文件读取数据源2
 * <p>
 * https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/datastream/sources/
 * https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/connectors/datastream/formats/csv/
 * <p>
 * csv 格式:
 * <p>
 * Berlin|52.5167|13.3833|Germany|DE|Berlin|primary|3644826
 * San Francisco|37.7562|-122.443|United States|US|California||3592294
 * Beijing|39.905|116.3914|China|CN|Beijing|primary|19433000
 * Beijing|Beijing|Beijing|China|CN|Beijing|primary <--错行
 * Beijing|39.905|116.3914|China|CN|Beijing|primary|19433000
 */
public class CsvFileTest02 {

    public static void main(String[] args) throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //加载数据源
        FileDataSource fileSource = new FileDataSource();
        FileSource<CityPojo> source = fileSource.getFileSource();
        DataStream dataStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "data/csv01");

        //转换
        SingleOutputStreamOperator<String> operator=dataStream.map(new MapFunction<CityPojo, String>() {
            @Override
            public String map(CityPojo cityPojo) throws Exception {
                return cityPojo.city;
            }
        });
        //sink
        operator.print();

        //执行
        env.execute();
    }
}
